package org.zhadoop.stream.cql.executor.operatorinfocreater;

import org.zhadoop.stream.api.ConfigAnnotation;
import org.zhadoop.stream.api.opereators.InnerInputSourceOperator;
import org.zhadoop.streaming.config.StreamingConfig;

public class KafkaInputOperator extends InnerInputSourceOperator{

	/**
	 * 分组Id
	 */
	@ConfigAnnotation(StreamingConfig.OPERATOR_KAFKA_GROUPID)
	private String groupId;
	
	/**
	 * topic
	 */
	@ConfigAnnotation(StreamingConfig.OPERATOR_KAFKA_TOPIC)
	private String topic;
	
	/**
     * kafka读取数据的zookeeper地址
     * 地址加端口，多个之间用逗号分隔
     * 比如：192.168.0.2:2181,192.168.0.3:2181
     */
    @ConfigAnnotation(StreamingConfig.OPERATOR_KAFKA_ZOOKEEPERS)
	private String zookeepers;
    
    /**
     * zookeeper连接超时时间
     */
    @ConfigAnnotation(StreamingConfig.OPERATOR_KAFKA_ZKSESSIONTIMEOUT)
    private Integer zkSessionTimeout;

    /**
     * kafka zk 同步时间参数
     */
    @ConfigAnnotation(StreamingConfig.OPERATOR_KAFKA_ZKSYNCTIME)
    private Integer zkSyncTime;
	
    /**
     * kafka消费者中获取到数据之后的序列化类
     */
    @ConfigAnnotation(StreamingConfig.OPERATOR_KAFKA_MESSAGESERIALIZERCLASS)
    private String messageSerializerClass;
    
    /**
     * 是否从头开始读取kafka数据
     * 默认false
     */
    @ConfigAnnotation(StreamingConfig.OPERATOR_KAFKA_READ_FROMBEGINNING)
    private Boolean fromBeginning;

	public KafkaInputOperator(String id, int parallelNumber) {
		super(id, parallelNumber);
	}

	public String getGroupId() {
		return groupId;
	}

	public void setGroupId(String groupId) {
		this.groupId = groupId;
	}

	public String getTopic() {
		return topic;
	}

	public void setTopic(String topic) {
		this.topic = topic;
	}

	public String getZookeepers() {
		return zookeepers;
	}

	public void setZookeepers(String zookeepers) {
		this.zookeepers = zookeepers;
	}

	public Integer getZkSessionTimeout() {
		return zkSessionTimeout;
	}

	public void setZkSessionTimeout(Integer zkSessionTimeout) {
		this.zkSessionTimeout = zkSessionTimeout;
	}

	public Integer getZkSyncTime() {
		return zkSyncTime;
	}

	public void setZkSyncTime(Integer zkSyncTime) {
		this.zkSyncTime = zkSyncTime;
	}

	public String getMessageSerializerClass() {
		return messageSerializerClass;
	}

	public void setMessageSerializerClass(String messageSerializerClass) {
		this.messageSerializerClass = messageSerializerClass;
	}

	public Boolean getFromBeginning() {
		return fromBeginning;
	}

	public void setFromBeginning(Boolean fromBeginning) {
		this.fromBeginning = fromBeginning;
	}
}
